package com.multi.thread;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;

/**
 * 使用循环栅栏（CyclicBarrier）实现多线程同步的示例。
 */
public class CyclicalBarrierDemo {

    public static void main(String[] args) throws Exception {
//        test0();
        test1();
//        test2();
    }

    public static void test0() {
        // 这里的 7 就是告诉 barrier 需要有7个线程调用 await()，它才会触发后面的回调（召唤神龙成功）。所以 barrier 是通过构造方法参数知道要等7个线程的。
        // CyclicBarrier 是一个同步辅助类，它允许一组线程互相等待，直到所有线程都到达某个公共屏障点。就是说，当所有线程都调用了 await() 方法后，才会继续执行后续的操作。
        // 用来控制线程的数量
        // 收集7棵龙珠，然后召唤神龙
        CyclicBarrier barrier = new CyclicBarrier(7, () -> System.out.println("召唤神龙成功！"));


        for (int i = 1; i <= 7; i++) {

            new Thread(() -> {
                try {
                    System.out.println("收集第 " + Thread.currentThread().getName() + " 棵龙珠");
                    barrier.await(); // 等待其他线程到达栅栏点
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }

            }, String.valueOf(i)).start();
        }
    }

    public static void test1() {
        // 创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 任务数量
        int taskCount = 5;

        // 创建CyclicBarrier，指定参与线程数和所有线程到达屏障后的回调
        CyclicBarrier barrier = new CyclicBarrier(taskCount, () -> {
            System.out.println("所有任务已完成，准备执行后续操作");
        });

        // 创建任务列表
        List<Future<String>> futures = new ArrayList<>();

        // 提交任务
        for (int i = 0; i < taskCount; i++) {
            final int taskId = i;
            Future<String> future = executor.submit(() -> {
                try {
                    // 模拟任务执行
                    System.out.println("任务 " + taskId + " 开始执行");
                    Thread.sleep((long) (Math.random() * 2000));
                    System.out.println("任务 " + taskId + " 执行完成");

                    // 等待所有任务完成
                    barrier.await();

                    return "任务 " + taskId + " 结果";
                } catch (InterruptedException | BrokenBarrierException e) {
                    Thread.currentThread().interrupt();
                    return "任务 " + taskId + " 被中断";
                }
            });
            futures.add(future);
        }

        try {
            // 收集结果
            for (Future<String> future : futures) {
                try {
                    System.out.println(future.get());
                } catch (ExecutionException e) {
                    System.out.println("任务执行异常: " + e.getMessage());
                }
            }

            System.out.println("\n主线程继续执行...");

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println("主线程等待被中断");
        } finally {
            // 关闭线程池
            executor.shutdown();
        }
    }


    /**
     *
     * @throws InterruptedException
     */
    public static void test2() throws InterruptedException {
        List<String> list = new ArrayList<>();

        Random random = new Random();
        for (int i = 0; i < 100000; i++) {
            list.add(String.valueOf(i) + random.nextInt(100));
        }

        final int threadCount = 10;
        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);

        // 每个线程处理的数据量
        int batchSize = list.size() / threadCount;
        CyclicBarrier barrier = new CyclicBarrier(threadCount, () -> {
            System.out.println("所有任务完成，主线程继续执行。" + Thread.currentThread().getName());
        });

        for (int i = 0; i < threadCount; i++) {
            int start = i * batchSize;
            int end = (i == threadCount - 1) ? list.size() : (i + 1) * batchSize;
            List<String> subList = list.subList(start, end);

            executorService.execute(() -> {
                for (String item : subList) {
                    // 模拟处理数据
                    /*try {
                        TimeUnit.MILLISECONDS.sleep(random.nextInt(50));
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }*/

                    System.out.println("处理数据: " + item + "，线程名：" + Thread.currentThread().getName());
                }
                try {
                    // cyclicalBarrier 是让所有的线程达到一个同步点，只有当所有线程都调用了 await() 方法后，才会继续执行后续的操作。
                    // 像这里将大量任务分批处理，肯定会有导致有的线程先完成，有的线程后完成的情况。那资源利用率就会不高。
                    // 可以使用CountDownLatch 或者 同步队列来实现更高效的线程同步。
                    barrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }

        executorService.shutdown();
        while (!executorService.awaitTermination(500, TimeUnit.MILLISECONDS)) {
            System.out.println("线程池未在500毫秒内完全终止，继续等待...");
        }

        System.out.println("我是主线程，我要接着往下执行了" + Thread.currentThread().getName());
    }

}
